package io.kiki.sba.registry.server.clientInterface.push;


import io.kiki.sba.registry.api.Callback;
import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.store.PushData;
import io.kiki.sba.registry.common.model.store.Watcher;
import io.kiki.sba.registry.core.model.ReceivedConfigData;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.node.service.ClientNodeService;
import io.kiki.sba.registry.task.MetricsableThreadPoolExecutor;
import io.kiki.sba.registry.task.RejectedDiscardHandler;
import io.kiki.sba.registry.util.OsUtils;
import io.kiki.stack.netty.exception.InvokeTimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

public class WatchProcessor {
    private static final Logger logger = LoggerFactory.getLogger(WatchProcessor.class);

    final RejectedDiscardHandler rejectedDiscardHandler = new RejectedDiscardHandler();
    private final ThreadPoolExecutor watchCallbackExecutor = MetricsableThreadPoolExecutor.newExecutor("WatchCallback", OsUtils.getCpuCount() * 2, 8000, rejectedDiscardHandler);
    @Autowired
    protected Config config;
    @Autowired
    protected PushSwitchService pushSwitchService;
    @Autowired
    protected PushDataGenerator pushDataGenerator;
    @Autowired
    protected ClientNodeService clientNodeService;

    boolean doExecuteOnWatch(Watcher watcher, ReceivedConfigData receivedConfigData, long triggerTimestamp) {
        if (!pushSwitchService.canIpPushLocal(watcher.getClientUrl().getIp())) {
            return false;
        }
        if (watcher.getPushedVersion() >= receivedConfigData.getVersion()) {
            return false;
        }
        PushData pushData = pushDataGenerator.createPushData(watcher, receivedConfigData);
        clientNodeService.pushWithCallback(pushData.getPayload(), watcher.getClientUrl(), new WatchPushCallback(triggerTimestamp, watcher, receivedConfigData.getVersion()));
        return true;
    }

    void trace(long triggerTimestamp, long pushStartTimestamp, Trace.PushStatus status, Watcher w, long version) {
        final long finishTs = System.currentTimeMillis();
        logger.info("{},{},{},{},delay={},{},regTs={},notifyTs={},addr={}", status, w.getDataInfoId(), version, w.getAppName(), finishTs - triggerTimestamp, finishTs - pushStartTimestamp, w.getRegisterTimestamp(), triggerTimestamp, w.getClientUrl().buildAddressString());
    }

    final class WatchPushCallback implements Callback {
        final long triggerTimestamp;
        final long pushStartTimestamp = System.currentTimeMillis();
        final Watcher watcher;
        final long version;

        WatchPushCallback(long triggerTimestamp, Watcher w, long version) {
            this.watcher = w;
            this.version = version;
            this.triggerTimestamp = triggerTimestamp;
        }

        @Override
        public void onSuccess(Channel channel, Object message) {
            watcher.updatePushedVersion(version);
            tracePush(Trace.PushStatus.OK);
        }

        @Override
        public void onFailed(Channel channel, Throwable exception) {
            if (exception instanceof InvokeTimeoutException) {
                tracePush(Trace.PushStatus.Timeout);
                logger.error("[PushTimeout]watcher={}, ver={}", watcher.shortDesc(), version);
            } else {
                final boolean channelConnected = channel.isConnected();
                if (channelConnected) {
                    tracePush(Trace.PushStatus.Fail);
                    logger.error("[PushFailed]watcher={}, ver={}", watcher.shortDesc(), version, exception);
                } else {
                    tracePush(Trace.PushStatus.ChanClosed);
                    logger.error("[PushChanClosed]watcher={}, ver={}", watcher.shortDesc(), version);
                }
            }
        }

        void tracePush(Trace.PushStatus status) {
            trace(triggerTimestamp, pushStartTimestamp, status, watcher, version);
        }

        @Override
        public Executor getExecutor() {
            return watchCallbackExecutor;
        }
    }
}
